package com.yi.wood.gob.gate.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.yi.wood.cache.DefaultChannelGroup;
import com.yi.wood.gob.gate.constant.HttpResp;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

import static io.netty.handler.codec.http.HttpUtil.isKeepAlive;
import static io.netty.handler.codec.http.HttpUtil.setContentLength;

/**
 * Created by moyi on 2018/9/20 10:22
 *
 * @author moyi
 */
@Service
@ChannelHandler.Sharable
public class GobHandler extends SimpleChannelInboundHandler {

    private static final Logger LOG = LoggerFactory.getLogger(GobHandler.class);

    private WebSocketServerHandshaker handShaker;

    @Resource
    private GobEventHandler gobEventHandler;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest) {
            // http请求
            handleHttpRequest(ctx, (FullHttpRequest)msg);
        } else {
            handleWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }

    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame msg) {
        //关闭链路指令
        if (msg instanceof CloseWebSocketFrame) {
            LOG.info("关闭连接");
        }
        //PING 消息
        else if (msg instanceof PingWebSocketFrame) {
            ctx.write(new PongWebSocketFrame(msg.content().retain()));
        }
        else if (!(msg instanceof TextWebSocketFrame)) {
            LOG.info("不支持此类报文");
        } else {
            //webSocket
            TextWebSocketFrame text = (TextWebSocketFrame) msg;
            String str = text.text();
            LOG.info("收到报文:{}", str);
            JSONObject obj = JSON.parseObject(str);
            // 处理消息
            gobEventHandler.execute(obj,ctx);

            Map<String, Object> map = new HashMap<>(2);
            map.put("key", "hello");
            String sendData = JSON.toJSONString(map);
            LOG.info("下发报文:{}",sendData);
            TextWebSocketFrame out = new TextWebSocketFrame(sendData);
            ctx.channel().writeAndFlush(out.retain());
        }
    }

    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest msg) {

        //HTTP 请异常
        if (!msg.decoderResult().isSuccess() || !HttpResp.WEB_SOCKET.equals(msg.headers().get(HttpResp.UPGRADE))) {
            System.out.println(msg.decoderResult().isSuccess());
            System.out.println(msg.headers().get(HttpResp.UPGRADE));
            sendHttpResponse(ctx, msg, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }

        // 建立连接握手
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://localhost:10001/websocket", null, false);
        handShaker = wsFactory.newHandshaker(msg);
        if (handShaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handShaker.handshake(ctx.channel(), msg);
        }
    }

    private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest msg, FullHttpResponse resp) {

        //响应
        if (resp.status().code() != HttpResp.SUCCESS) {
            ByteBuf buf = Unpooled.copiedBuffer(resp.status().toString(), CharsetUtil.UTF_8);
            resp.content().writeBytes(buf);
            buf.release();
            setContentLength(resp, resp.content().readableBytes());
        }

        //非Keep-Alive,关闭链接
        ChannelFuture future = ctx.channel().writeAndFlush(resp);
        if (!isKeepAlive(resp) || resp.status().code() != HttpResp.SUCCESS) {
            future.addListener(ChannelFutureListener.CLOSE);
        }


    }
    /**
     * 上线
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        LOG.info("↑一个连接上线:{}", ctx.channel().remoteAddress().toString());
        DefaultChannelGroup.getChannels().add(ctx.channel());
        super.channelActive(ctx);
    }

    /**
     * 下线
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        LOG.info("↓一个连接下线:{}", ctx.channel().remoteAddress().toString());
        DefaultChannelGroup.getChannels().remove(ctx);
        super.channelInactive(ctx);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        LOG.info("↓一个连接异常:{}，cause:{}", ctx.channel().remoteAddress().toString(),cause);
        ctx.close();
        ctx.channel().close();
    }
}
